package com.wing.util;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.nacos.common.utils.JacksonUtils;
import com.wing.config.PulsarClientInit;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.TypedMessageBuilder;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

@Component
public class PulsarUtils {

    static PulsarClientInit pulsarclientinitultimate =null;

    public PulsarUtils(PulsarClientInit pulsarclientinitultimate) {
        PulsarUtils.pulsarclientinitultimate = pulsarclientinitultimate;
    }

    /**
     * 初始化Producer
     * @param topic 主题
     * @return producer
     * @throws PulsarClientException
     */
    private static TypedMessageBuilder initProducer(String topic) throws PulsarClientException {
        PulsarClient client = pulsarclientinitultimate.getPulsars();

        return  client.newProducer()
                .topic(topic)
                .create()
                .newMessage(Schema.BYTES);
    }

    /**
     * 异步发送
     * @param topic 主题t
     * @param message 要发送的数据
     */
    public static <T> void sendAnsyMsg(String topic,  T message) throws PulsarClientException {
        TypedMessageBuilder builder = initProducer(topic);
        CompletableFuture future = builder.value(JSONObject.toJSONBytes(message)).sendAsync();//异步发送
        future.handle((v, ex) -> {
            if (ex == null) {
                System.out.println("Message persisted2: {}"+message);
            } else {
                System.out.println("发送Pulsar消息失败msg:【{}】 "+message+ex);
            }
            return null;
        });
        // future.join();
    }

    /**
     * 延迟发送消息
     * @param topic    主题
     * @param message  消息
     * @param catchEx  是否捕获异常
     * @param delay    延迟时间
     * @param timeUnit 单位
     * @param <T>      消息类型
     * @return mq消息ID
     */
    public static <T> String send(String topic, T message, boolean catchEx, Long delay, TimeUnit timeUnit) throws PulsarClientException {
        TypedMessageBuilder<byte[]> builder = initProducer(topic);
        if (ObjectUtil.isAllNotEmpty(delay, timeUnit)) {
            builder.deliverAfter(delay, timeUnit);
        }
        String msg = message instanceof Number ? message.toString() : JSONUtil.toJsonStr(message);
        builder.value(JacksonUtils.toJsonBytes(msg));
        try {
            System.out.println("发送MQ消息topic:{}，message:{}"+ topic+ "---"+msg);
            return builder.send().toString();
        } catch (Exception e) {
            System.out.println("发送topic:{}消息失败，message:{}"+topic+msg);
            if (!catchEx) {
                throw new RuntimeException("mq发送失败");
            }
            return null;
        }
    }

    /**
     * 简单发送需要捕获异常
     * @param topic 主题
     * @param message 信息
     * @param catchEx 是否捕获异常
     * @param <T> 消息类型
     * @return
     * @throws PulsarClientException
     */
    public static <T> String send(String topic, T message, boolean catchEx) throws PulsarClientException {
        TypedMessageBuilder<byte[]> builder = initProducer(topic);
        return send(topic, message, catchEx, null, null);
    }

    /**
     * 延迟发送消息
     * @param topic    主题
     * @param message  消息
     * @param delay    延迟时间
     * @param timeUnit 单位
     * @param <T>      消息类型
     * @return mq消息ID
     */

    public static <T> String send(String topic, T message, Long delay, TimeUnit timeUnit) throws PulsarClientException {
        return send(topic, message, true, delay, timeUnit);
    }

    /**
     * 简单发送
     * @param topic 主题
     * @param message 消息
     * @param <T> 类型
     * @return
     * @throws PulsarClientException
     */
    public static <T> String send(String topic, T message) throws PulsarClientException {
        return send(topic, message, true,null,null);
    }

}
